package cn.gemiman.daoyi.module.iot.open.jhl.kafka.config;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

/**
 * 道一kafka配置
 *
 */
@Configuration
public class DaoyiKafkaConfig {

    /**
     * 读取道一kafka配置
     * Primary注解表示默认以这个为准
     *
     * @return 道一kafka配置
     */
    @Primary
    @ConfigurationProperties(prefix = "spring.kafka.daoyi")
    @Bean
    public KafkaProperties daoyiKafkaProperties() {
        return new KafkaProperties();
    }

    /**
     * 构建道一kafka的生产者发送template
     *
     * @param daoyiKafkaProperties 道一kafka配置
     * @return 道一kafka的生产者发送template
     */
    @Primary
    @Bean
    public KafkaTemplate<Object, Object> daoyiKafkaTemplate(
            @Autowired @Qualifier("daoyiKafkaProperties") KafkaProperties daoyiKafkaProperties) {
        return new KafkaTemplate<>(daoyiProducerFactory(daoyiKafkaProperties));
    }

    /**
     * 构建道一kafka的消费者监听容器工厂
     *
     * @param daoyiKafkaProperties 道一kafka配置
     * @return 道一kafka的消费者监听容器工厂
     */
    @Bean("daoyiKafkaListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
    daoyiKafkaListenerContainerFactory(@Autowired @Qualifier("daoyiKafkaProperties") KafkaProperties daoyiKafkaProperties) {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(daoyiConsumerFactory(daoyiKafkaProperties));
        return factory;
    }

    /**
     * 新建道一kafka的消费者工厂
     *
     * @param daoyiKafkaProperties 道一kafka配置
     * @return 道一kafka的消费者工厂
     */
    private ConsumerFactory<? super Integer, ? super String> daoyiConsumerFactory(KafkaProperties daoyiKafkaProperties) {
        return new DefaultKafkaConsumerFactory<>(daoyiKafkaProperties.buildConsumerProperties(null));
    }

    /**
     * 新建道一kafka的生产者工厂
     *
     * @param daoyiKafkaProperties 道一kafka配置
     * @return 道一kafka的生产者工厂
     */
    private DefaultKafkaProducerFactory<Object, Object> daoyiProducerFactory(KafkaProperties daoyiKafkaProperties) {
        return new DefaultKafkaProducerFactory<>(daoyiKafkaProperties.buildProducerProperties(null));
    }
}

